package org.lib.kafkahello.try2;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class Try2Config {

	/**
	 * 主题02：topic02，拥有2个分区
	 */
	public static final String TOPIC_02 = "topic02";
	
	/**
	 * 主题topic02分区数：2
	 */
	public static final int TOPIC_02_P_NUMBER = 2;
	
	/**
	 * 主题02-KEY：topic02_key
	 */
	public static final String TOPIC_02_KEY = "topic02_key";
	
	public static final String TRY2_GROUP_ID_01 = "try2group01";
	public static final String TRY2_GROUP_ID_02 = "try2group02";

	// groupId = TRY2_GROUP_ID_01
	// 消费者1
	@KafkaListener(id="try2listenerA", topics = {TOPIC_02}, groupId = TRY2_GROUP_ID_01)
	public void listenTopic1(ConsumerRecord<?, ?> record) {
		log.info("try2-消费-A：topic={}, partition={}, offset={}, key=[{}], timestamp={}, value=[{}]",
				record.topic(), record.partition(), record.offset(), record.key(), record.timestamp(), record.value());
	}
	
	// groupId = TRY2_GROUP_ID_01
	// 消费者2
	@KafkaListener(id="try2listenerB", topics = {TOPIC_02}, groupId = TRY2_GROUP_ID_01)
	public void listenTopic2(ConsumerRecord<?, ?> record) {
		log.info("try2-消费-B：topic={}, partition={}, offset={}, key=[{}], timestamp={}, value=[{}]",
				record.topic(), record.partition(), record.offset(), record.key(), record.timestamp(), record.value());
	}
	
	// groupId = TRY2_GROUP_ID_01
	// 消费者3：超过分区数量了
//	@KafkaListener(id="try2listenerC", topics = {TOPIC_02}, groupId = TRY2_GROUP_ID_01)
//	public void listenTopic3(ConsumerRecord<?, ?> record) {
//		log.info("try2-消费-C：topic={}, partition={}, offset={}, key=[{}], timestamp={}, value=[{}]",
//				record.topic(), record.partition(), record.offset(), record.key(), record.timestamp(), record.value());
//	}
	
	// ---------------------
	
	// 测试多个消费者组
	
//	// groupId = TRY2_GROUP_ID_02
//	// 消费者4
//	// 注意：id开头的 g2 在 日志中 的线程名 里面消失了
//	@KafkaListener(id="try2g2listenerH", topics = {TOPIC_02}, groupId = TRY2_GROUP_ID_02)
//	public void listenTopic4(ConsumerRecord<?, ?> record) {
//		log.info("try2-消费-H：topic={}, partition={}, offset={}, key=[{}], timestamp={}, value=[{}]",
//				record.topic(), record.partition(), record.offset(), record.key(), record.timestamp(), record.value());
//	}
//	
//	// groupId = TRY2_GROUP_ID_02
//	// 消费者5
//	@KafkaListener(id="try2g2listenerJ", topics = {TOPIC_02}, groupId = TRY2_GROUP_ID_02)
//	public void listenTopic5(ConsumerRecord<?, ?> record) {
//		log.info("try2-消费-I：topic={}, partition={}, offset={}, key=[{}], timestamp={}, value=[{}]",
//				record.topic(), record.partition(), record.offset(), record.key(), record.timestamp(), record.value());
//	}
	
}
